package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/11/06
 * 版本渠道地区访客类别聚合统计
 * 需要启动的进程
 *      zk、kafka、flume、BaseLogSplit、DwsTrafficVcChArIsNewPageViewWindow
 */
public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(
                10022,
                4,
                "dws_traffic_vc_ch_ar_is_new_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对流中数据进行类型转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS
                = kafkaStrDS.map(JSON::parseObject);
        //jsonObjDS.print();
        //TODO 2.按照设备的id进行分组
        KeyedStream<JSONObject, String> midKeyedDS 
                = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        //TODO 3.对分组后的数据进行处理   jsonObj->统计的实体类对象  相当于WordCount封装二元组   使用Flink的状态编程，判断是否为当日独立访客
        /*关于Flink的状态编程
        *   状态：用于存储程序运行的中间结果
        *   状态和检查点之间的关系：  检查点是对状态的备份  可以理解为检查点是状态的一个副本
        *   状态分类
        *       原始状态
        *           由程序员关联状态的存储
        *       托管状态
        *           由Flink框架对状态进行管理
        *           算子状态
        *               作用范围：算子的每一个并行子任务
        *               List、广播
        *               初始化：  CheckpointedFunction->initializeState->context.getXXXState(状态描述器对象)
        *           键控状态
        *               作用范围：经过keyby之后的每一个组
        *               ValueState、ListState、MapState、ReducingState、AggregatingState
        *               初始化：在open方法中对状态进行初始化     getRuntimeContext.getXXXState(状态描述器对象)
        */
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = midKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                    ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor
                                = new ValueStateDescriptor<String>("lastVisitDateState", String.class);
                        //设置状态的保留时间
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        //获取当前设备上次访问日期
                        String lastVisitDate = lastVisitDateState.value();
                        //获取当前设备这次访问日期
                        Long ts = jsonObj.getLong("ts");
                        String curVisitDate = DateFormatUtil.tsToDate(ts);
                        Long uvCt = 0L;

                        if (StringUtils.isEmpty(lastVisitDate) || !curVisitDate.equals(lastVisitDate)) {
                            uvCt = 1L;
                            lastVisitDateState.update(curVisitDate);
                        }

                        Long svCt = StringUtils.isEmpty(pageJsonObj.getString("last_page_id")) ? 1L : 0L;

                        TrafficPageViewBean viewBean = new TrafficPageViewBean(
                                "",
                                "",
                                "",
                                commonJsonObj.getString("vc"),
                                commonJsonObj.getString("ch"),
                                commonJsonObj.getString("ar"),
                                commonJsonObj.getString("is_new"),
                                uvCt,
                                svCt,
                                1L,
                                pageJsonObj.getLong("during_time"),
                                ts
                        );
                        out.collect(viewBean);
                    }
                }
        );
        //beanDS.print();
        //TODO 4.指定Watermark以及提取事件时间字段
        /**
         * 关于Flink的水位线
         *      前提：事件时间语义
         *      是Flink内部的逻辑时钟
         *      用于衡量事件时间进展的标记
         *      主要用于窗口触发计算、窗口关闭以及定时器的执行
         *      也会作为流中的元素，随着流的流动进行传递
         *      是递增的，永远不会减小
         *      FlinkAPI提供了两种默认的水位线生成策略
         *          单调递增-forMonotonousTimestamps
         *          有界乱序-forBoundedOutOfOrderness(乱序程度)
         *          单调递增是有界乱序的子类  ，单调递增是有界乱序的特殊情况(乱序程度是0)
         *          底层实现了 WatermarkGenerator
         *              onEvent: 流中每条元素到来的时候都会执行的方法---维护流中元素最大时间
         *              onPeriodicEmit:周期性执行的方法(默认周期200ms)---创建Watermark对象  最大时间-乱序程序-1ms
         *      水位线的传递
         *          上游1个并行度，下游n个并行度---广播
         *          上游n个并行度，下游1个并行度---将上游所有并行度上的Watermark拿到，取最小
         *          上游n个并行度，下游n个并行度---先广播，再取最小
         */

        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TrafficPageViewBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TrafficPageViewBean>() {
                                    @Override
                                    public long extractTimestamp(TrafficPageViewBean bean, long recordTimestamp) {
                                        return bean.getTs();
                                    }
                                }
                        )
        );
        //TODO 5.按照统计的维度进行分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> dimKeyedDS = withWatermarkDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean bean) throws Exception {
                return Tuple4.of(
                        bean.getVc(),
                        bean.getCh(),
                        bean.getAr(),
                        bean.getIsNew()
                );
            }
        });

        //TODO 6.开窗
        /**
         * 关于FlinkAPI中的窗口
         *      定义：将无限数据流划分为有限的数据库进行处理
         *      分类
         *          按照驱动形式分
         *              时间窗口
         *              计数窗
         *          按照数据划分方式分
         *              滚动窗口
         *              滑动窗口
         *              会话窗口
         *              全局窗口
         *      开窗前是否进行了keyBy
         *          keyBy:  针对keyBy后的每一个组独立开窗，窗口之间相互不影响
         *          no-keyBy:  针对整条流进行开窗，并行度强制设置为1
         *      以滚动事件时间窗口为例
         *          窗口对象什么时候创建
         *              当属于这个窗口的第一个元素到来的时候，创建窗口对象
         *          窗口是左闭右开的
         *          窗口的起始时间
         *              向下取整
         *          窗口的结束时间
         *              起始时间 + 窗口大小
         *          窗口的最大时间
         *              结束时间 - 1ms
         *          窗口什么时候触发计算
         *              window.maxTimestamp() <= ctx.getCurrentWatermark()
         *          窗口什么关闭
         *              Watermark到了window.maxTimestamp() + allowedLateness
         *      迟到数据如何处理
         *          在指定Watermark的时候设置乱序程序
         *          在开窗的时候设置窗口的允许迟到时间
         *          侧输出流
         */
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS
                = dimKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 7.聚合
        /**
         * 关于FlinkAPI的处理函数
         *      增量聚合
         *          优势：数据来一条处理一次，不会缓存所有数据，节省空间   劣势：获取不到窗口详细信息
         *          reduce
         *              窗口中元素类型和向下游传递的类型一致
         *          aggregate
         *              窗口中元素类型、累加器类型以及向下游传递的数据类型可以不一致
         *      全量聚合
         *          优势：可以获取到窗口详细信息     劣势：会缓存所有数据，费空间
         *          apply
         *              参数接收的是窗口对象
         *          process
         *              参数接收的是上下文对象，除了可以获取窗口对象之外，还可以获取水位线相关信息   更底层
         *      增量 + 全量
         */
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                },
                /*new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {

                    }
                }*/
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<String, String, String, String> stringStringStringStringTuple4, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>.Context context, Iterable<TrafficPageViewBean> elements, Collector<TrafficPageViewBean> out) throws Exception {
                        TimeWindow window = context.window();
                        TrafficPageViewBean viewBean = elements.iterator().next();

                        viewBean.setStt(DateFormatUtil.tsToDateTime(window.getStart()));
                        viewBean.setEdt(DateFormatUtil.tsToDateTime(window.getEnd()));
                        viewBean.setCur_date(DateFormatUtil.tsToDate(window.getStart()));
                        out.collect(viewBean);
                    }
                }
        );
        //TODO 8.将聚合的结果写到Doris
        reduceDS.print();
        reduceDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));

    }
}
